package com.lhc.kafkademo.producer;

import org.apache.kafka.clients.producer.*;

import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

/**
 * Kafka 生产api
 */
public class ProducerSample {

    public final static String TOPIC_NAME="jz_topic";

    public static void main(String[] args) throws ExecutionException, InterruptedException {

        //producerSend();

        //异步阻塞发送
//        producerSendSync();

//        //异步发送带回调函数
//        producerSendWithCallBack();

        producerSendWithCallBackAndPartition();
    }

    /**
     * producer 异步发送
     */
    public static void producerSend(){
        Properties properties = new Properties();
        properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.11.130:9092");
        properties.setProperty(ProducerConfig.ACKS_CONFIG,"all");
        properties.setProperty(ProducerConfig.RETRIES_CONFIG,"0");
        properties.setProperty(ProducerConfig.BATCH_SIZE_CONFIG,"16384");
        properties.setProperty(ProducerConfig.LINGER_MS_CONFIG,"1");
        properties.setProperty(ProducerConfig.BUFFER_MEMORY_CONFIG,"33554432");

        properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
        properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");

        //Producer 主对象
        KafkaProducer<String, String> producer = new KafkaProducer<>(properties);

        //消息对象 - ProducerRecord
        for (int i = 0; i < 10; i++) {
            ProducerRecord<String , String> record =
                    new ProducerRecord<String, String>(TOPIC_NAME,"key-"+i,"value-"+i);
            //发送消息
            Future<RecordMetadata> send = producer.send(record);
        }
        //关闭发送通道
        producer.close();
    }

    /**
     * 同步发送，异步阻塞发送
     */
    public static void producerSendSync() throws ExecutionException, InterruptedException {
        Properties properties = new Properties();
        properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.11.130:9092");
        properties.setProperty(ProducerConfig.ACKS_CONFIG,"all");
        properties.setProperty(ProducerConfig.RETRIES_CONFIG,"0");
        properties.setProperty(ProducerConfig.BATCH_SIZE_CONFIG,"16384");
        properties.setProperty(ProducerConfig.LINGER_MS_CONFIG,"1");
        properties.setProperty(ProducerConfig.BUFFER_MEMORY_CONFIG,"33554432");

        properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
        properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");

        //Producer 主对象
        KafkaProducer<String, String> producer = new KafkaProducer<>(properties);

        //消息对象 - ProducerRecord
        for (int i = 0; i < 10; i++) {
            String key = "key-"+i;
            ProducerRecord<String , String> record =
                    new ProducerRecord<String, String>(TOPIC_NAME,key,"value-"+i);
            //发送消息
            Future<RecordMetadata> send = producer.send(record);
            RecordMetadata recordMetadata = send.get();
            System.out.println("key:"+key+",partition:"+recordMetadata.partition()
                    +",offset:"+recordMetadata.offset()
            );
        }
        //关闭发送通道
        producer.close();
    }

    /**
     * 异步发送带回调
     * @throws ExecutionException
     * @throws InterruptedException
     */
    public static void producerSendWithCallBack() throws ExecutionException, InterruptedException {
        Properties properties = new Properties();
        properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.11.130:9092");
        properties.setProperty(ProducerConfig.ACKS_CONFIG,"all");
        properties.setProperty(ProducerConfig.RETRIES_CONFIG,"0");
        properties.setProperty(ProducerConfig.BATCH_SIZE_CONFIG,"16384");
        properties.setProperty(ProducerConfig.LINGER_MS_CONFIG,"1");
        properties.setProperty(ProducerConfig.BUFFER_MEMORY_CONFIG,"33554432");

        properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
        properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");

        //Producer 主对象
        KafkaProducer<String, String> producer = new KafkaProducer<>(properties);

        //消息对象 - ProducerRecord
        for (int i = 0; i < 10; i++) {
            String key = "key-"+i;
            ProducerRecord<String , String> record =
                    new ProducerRecord<String, String>(TOPIC_NAME,key,"value-"+i);
            //发送消息
            Future<RecordMetadata> send = producer.send(record, new Callback() {
                @Override
                public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                    System.out.println("key:"+key+",partition:"+recordMetadata.partition()
                            +",offset:"+recordMetadata.offset()
                    );
                }
            });
        }
        //关闭发送通道
        producer.close();
    }

    /**
     * 异步发送带回调和partition负载均衡
     */
    public static void producerSendWithCallBackAndPartition() {
        Properties properties = new Properties();
        properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.11.130:9092");
        properties.setProperty(ProducerConfig.ACKS_CONFIG,"all");
        properties.setProperty(ProducerConfig.RETRIES_CONFIG,"0");
        properties.setProperty(ProducerConfig.BATCH_SIZE_CONFIG,"16384");
        properties.setProperty(ProducerConfig.LINGER_MS_CONFIG,"1");
        properties.setProperty(ProducerConfig.BUFFER_MEMORY_CONFIG,"33554432");

        properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
        properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
        properties.setProperty(ProducerConfig.PARTITIONER_CLASS_CONFIG,"com.lhc.kafkademo.producer.SamplePartition");

        //Producer 主对象
        KafkaProducer<String, String> producer = new KafkaProducer<>(properties);

        //消息对象 - ProducerRecord
        for (int i = 0; i < 10; i++) {
            String key = "key-"+i;
            ProducerRecord<String , String> record =
                    new ProducerRecord<String, String>(TOPIC_NAME, key,"value-"+i);
            //发送消息
            producer.send(record, new Callback() {
                @Override
                public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                    System.out.println("key:"+key+",partition:"+recordMetadata.partition()
                            +",offset:"+recordMetadata.offset()
                    );
                }
            });
        }
        //关闭发送通道
        producer.close();
    }
}
